package com.sh.data.engine.domain.datadev.offline.job;

import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.sh.data.engine.common.constants.Constants.RedisKey;
import com.sh.data.engine.common.enumDefinition.DSType;
import com.sh.data.engine.domain.base.model.enums.Database;
import com.sh.data.engine.domain.datadev.offline.model.domain.OfflineDevTaskTableInfoDomain;
import com.sh.data.engine.domain.integration.datasource.model.domain.DataSourceDetailDomain;
import com.sh.data.engine.domain.integration.datasource.service.DataSourceService;
import com.sh.data.engine.domain.shims.DbManagerFactory;
import com.sh.data.engine.domain.shims.db.BaseDbManager;
import com.sh.data.engine.domain.shims.db.DbOptions;
import com.sh.data.engine.domain.shims.db.model.FieldInfoDomain;
import com.sh.data.engine.domain.shims.db.model.TableInfoDomain;
import com.sh.data.engine.domain.workspace.index.model.domain.ProjectDomain;
import com.sh.data.engine.domain.workspace.index.service.ProjectService;
import com.sh.data.engine.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.sql.SQLException;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;

@Component
@Slf4j
public class OfflineDatadevSqlTask {

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private ProjectService projectService;

    @Autowired
    private DataSourceService dataSourceService;

    @XxlJob(value = "sqlKeywordLeader")
    public void sqlKeywordLeader() throws SQLException {

        List<ProjectDomain> allProject = projectService.getAllProject();
        for (ProjectDomain projectDomain : allProject) {
            log.info("projectId is {} ,name is{}", projectDomain.getId(), projectDomain.getName());
            DataSourceDetailDomain dataSourceDetail =
                dataSourceService.getDataSourceDetailByProjectId(projectDomain.getId(), false);

            if (Objects.isNull(dataSourceDetail)) {
                continue;
            }
            String hiveSiteAddress = null;

            String schema = null;
            if (DSType.Hive.name().equals(dataSourceDetail.getDsType())) {
                hiveSiteAddress = dataSourceDetail.getHiveConfig().getHiveSiteAddress();
            }

            if (DSType.isRdbms(dataSourceDetail.getDsType())) {
                schema = dataSourceDetail.getRdbmsConfig().getSchema();
            }
            Database from = Database.from(dataSourceDetail.getDsType());
            DbOptions dbOptions =
                new DbOptions(
                    from,
                    dataSourceDetail.getUsername(),
                    dataSourceDetail.getPassword(),
                    dataSourceDetail.getDsLink(),
                    hiveSiteAddress);
            dbOptions.setSchema(schema);
            BaseDbManager dbManager = null;
            try {
                dbManager = DbManagerFactory.getDbManager(from, dbOptions);
            } catch (Exception e) {
                log.info(
                    "projectId is skip{} ,name is skip {}", projectDomain.getId(), projectDomain.getName());
                continue;
            }
            List<TableInfoDomain> tableAndViewList = Lists.newArrayList();

            try {
                tableAndViewList = dbManager.getTableAndViewList(dataSourceDetail.getDbName(), schema);
            } catch (Exception e) {
                log.info(
                    "projectId is skip{} ,name is skip {}", projectDomain.getId(), projectDomain.getName());
                dbManager.close();
                continue;
            }
            List<String> tables =
                tableAndViewList.stream().map(e -> e.getTableName()).collect(Collectors.toList());
            OfflineDevTaskTableInfoDomain offlineDevTaskTableInfoDomain =
                new OfflineDevTaskTableInfoDomain();
            offlineDevTaskTableInfoDomain.setTables(tables);

            Set<String> fieldSet = Sets.newHashSet();
            for (TableInfoDomain table : tableAndViewList) {
                try {
                    List<FieldInfoDomain> fieldList =
                        dbManager.getFieldList(table.getDbName(), table.getSchema(), table.getTableName());

                    Set<String> fieldSets =
                        fieldList.stream().map(e -> e.getFieldName()).collect(Collectors.toSet());
                    fieldSet.addAll(fieldSets);
                } catch (Exception e) {
                    log.error("get field error ,skip table {}", table.getTableName());
                    continue;
                }
            }
            offlineDevTaskTableInfoDomain.setTableColumns(fieldSet);
            String sqlKeyword = String.format(RedisKey.DATADEV_SQL_KEYWORD, projectDomain.getId());
            String value = JSONObject.toJSONString(offlineDevTaskTableInfoDomain);
            redisTemplate.opsForValue().set(sqlKeyword, value);
            if (DSType.Hive.name().equals(dataSourceDetail.getDsType())) {
                dbManager.close();
            }
        }
    }
}
